"""
Case Type   : 防篡改
Case Name   : 账本数据库基础语法适配(DDL/DML)--enable_copy_server_files打开,不同权限用户使用copyfrom命令导入防篡改用户表
Create At   : 2022-02-28
Owner       : lwx1080719
Description :
    1.查询三权分立参数
    2.创建不同权限用户
    3.系统管理员创建schema并指定防篡改选项
    4.普通用户登录数据库在防篡改schema下创建防篡改用户表
    5.修改enable_copy_server_files为on
    6.系统管理员copyto命令导出防篡改用户表
    7.查询导出的文件是否存在
    8.不同权限用户登录数据库使用copyfrom命令导入防篡改用户表
    9.清理环境
Expect      :
    1.显示默认值off
    2.成功
    3.成功
    4.成功
    5.成功
    6.成功
    7.成功
    8.系统管理员导入成功，其他用户失败
    9.成功
History     :
"""




import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Constant import Constant
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Logger import Logger


class ModifyCase(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f'-----{os.path.basename(__file__)} start-----')
        self.dbuserNode = Node('PrimaryDbUser')
        self.primary_sh = CommonSH('PrimaryDbUser')
        self.Constant = Constant()
        self.Common = Common()
        self.user_path = os.path.join(macro.DB_INSTANCE_PATH, 'user.bat')
        self.schema_name = 's_security_tamper_proof_0052'
        self.table_name = 't_security_tamper_proof_0052'
        self.user_name = 'u_security_tamper_proof'
        self.user_privileges = {f'{self.user_name}_sys': 'sysadmin',
                                f'{self.user_name}_create': 'createrole',
                                f'{self.user_name}_audit': 'auditadmin',
                                f'{self.user_name}_noruser': '',
                                f'{self.user_name}_independent': 'independent',
                                f'{self.user_name}_mon': 'monadmin',
                                f'{self.user_name}_opr': 'opradmin',
                                f'{self.user_name}_pol': 'poladmin',
                                f'{self.user_name}_persistent': 'persistence'}
        self.default_value1 = self.Common.show_param('enableSeparationOfDuty')
        self.default_value = self.Common.show_param('enable_copy_server_files')

    def test_security(self):
        text = '----step1:查询参数值  expect:默认值off----'
        self.logger.info(text)
        self.assertEqual("off", self.default_value1, "执行失败:" + text)
        self.assertEqual("off", self.default_value, "执行失败:" + text)

        text = '----step2:创建不同权限用户 expect:成功----'
        self.logger.info(text)
        for user, privilege in self.user_privileges.items():
            create_cmd = f"drop user if exists {user} cascade; " \
                f"create user {user} {privilege} " \
                f"password '{macro.PASSWD_REPLACE}' ; "
            msg = self.primary_sh.execut_db_sql(create_cmd)
            self.logger.info(msg)
            self.assertIn(self.Constant.DROP_ROLE_SUCCESS_MSG and
                          self.Constant.CREATE_ROLE_SUCCESS_MSG, msg,
                          '执行失败:' + text)

        text = '----step3:系统管理员创建schema并指定防篡改选项 expect:成功----'
        self.logger.info(text)
        sql_cmd = f'''
            create schema {self.schema_name} authorization
            {self.user_name}_noruser with blockchain;'''
        self.logger.info(sql_cmd)
        msg = self.primary_sh.execut_db_sql(
            sql_cmd, sql_type=f' -U {self.user_name}_sys '
            f'-W {macro.PASSWD_REPLACE}')
        self.logger.info(msg)
        self.assertTrue(
            msg.count(self.Constant.CREATE_SCHEMA_SUCCESS_MSG) == 1,
            "执行失败:" + text)

        text = '----step4:普通用户登录数据库在防篡改schema下' \
               '创建防篡改用户表 expect:成功----'
        self.logger.info(text)
        sql_cmd = f'''
            create table {self.schema_name}.{self.table_name}
            (id int primary key ,name varchar(20) NOT NULL ,
            address text DEFAULT 'xian');'''
        self.logger.info(sql_cmd)
        msg = self.primary_sh.execut_db_sql(
            sql_cmd, sql_type=f' -U {self.user_name}_noruser '
            f'-W {macro.PASSWD_REPLACE}')
        self.logger.info(msg)
        self.assertIn(self.Constant.CREATE_TABLE_SUCCESS, msg,
                      "执行失败:" + text)

        text = '----step5:修改enable_copy_server_files为on expect:成功----'
        self.logger.info(text)
        sql_cmd = self.primary_sh.execut_db_sql(f'''
                alter system set enable_copy_server_files to on;
                select pg_sleep(2);
                show enable_copy_server_files;''')
        self.logger.info(sql_cmd)
        self.assertEqual("on", sql_cmd.split("\n")[-2].strip(),
                         "执行失败:" + text)

        text = '----step6:系统管理员copyto命令导出防篡改用户表 ' \
               'expect:成功----'
        self.logger.info(text)
        sql_cmd = f'''copy {self.schema_name}.{self.table_name} to 
                    '{self.user_path}';'''
        self.logger.info(sql_cmd)
        msg = self.primary_sh.execut_db_sql(
            sql_cmd, sql_type=f' -U {self.user_name}_sys '
            f'-W {macro.PASSWD_REPLACE}')
        self.logger.info(msg)
        self.assertTrue('COPY', "执行失败:" + text)

        text = '----step7:查询导出的防篡改用户表 expect:成功----'
        self.logger.info(text)
        cmd = f"ls {macro.DB_INSTANCE_PATH} | grep 'user.bat'"
        self.logger.info(cmd)
        result = self.Common.get_sh_result(self.dbuserNode, cmd)
        self.assertIn('user.bat', result, "执行失败" + text)

        text = '----step8:不同权限用户登录数据库使用' \
               'copyfrom命令导入防篡改用户表 ' \
               'expect:系统管理员导入成功，其他用户失败----'
        self.logger.info(text)
        sql_cmd = f'''copy {self.schema_name}.{self.table_name} 
                from '{self.user_path}';'''
        self.logger.info(sql_cmd)
        for user in self.user_privileges.keys():
            msg = self.primary_sh.execut_db_sql(
                sql_cmd, sql_type=f' -U {user} -W {macro.PASSWD_REPLACE}')
            self.logger.info(msg)
            if user in {f'{self.user_name}_sys'}:
                self.assertIn('COPY', msg, '执行失败:' + text)
            else:
                self.assertIn('must be system admin or a member', msg,
                          "执行失败:" + text)

    def tearDown(self):
        text = '----step9:清理环境 expect:成功----'
        self.logger.info(text)
        sql_cmd = self.primary_sh.execut_db_sql(f'''
            drop table  {self.schema_name}.{self.table_name} cascade;
            drop schema {self.schema_name} cascade;''')
        self.logger.info(sql_cmd)
        self.assertIn(self.Constant.DROP_SCHEMA_SUCCESS_MSG and
                      self.Constant.TABLE_DROP_SUCCESS, sql_cmd,
                      "执行失败:" + text)
        text = '----删除用户 expect:成功----'
        self.logger.info(text)
        drop_msg = []
        for user in self.user_privileges.keys():
            drop_cmd = f"drop user if exists {user} cascade; "
            msg = self.primary_sh.execut_db_sql(drop_cmd)
            self.logger.info(msg)
            drop_msg.append(msg)
        self.assertEqual(9,
                         drop_msg.count(self.Constant.DROP_ROLE_SUCCESS_MSG),
                         '执行失败:' + text)
        text = '-----恢复参数enable_copy_server_files默认值-----'
        self.logger.info(text)
        sql_cmd = self.primary_sh.execut_db_sql(f'''
            alter system set enable_copy_server_files to {self.default_value};
            select pg_sleep(2);
            show enable_copy_server_files;''')
        self.logger.info(sql_cmd)
        self.assertEqual("off", sql_cmd.split("\n")[-2].strip(),
                         "执行失败:" + text)
        text = '-----删除文件-----'
        self.logger.info(text)
        del_file = f'''rm -rf {self.user_path}'''
        self.logger.info(del_file)
        self.Common.get_sh_result(self.dbuserNode, del_file)
        cmd = f'if [ -f {self.user_path} ]; then echo "does exists"; ' \
                f'else echo "not exists"; fi'
        self.logger.info(del_file)
        file_judgments = self.Common.get_sh_result(self.dbuserNode, cmd)
        self.assertIn('not exists', file_judgments, "执行失败" + text)
        self.logger.info(f'-----{os.path.basename(__file__)} end-----')
